package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.WarnBaseBean;
import cn.itcast.inter.ProcessDataCepInterface;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.mail.MailSend;
import cn.itcast.util.RedisUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
import java.util.Map;

//实时预警，涨跌幅=(期末收盘点位-期初前收盘点位)/期初前收盘点位*100%
public class UpdownTask implements ProcessDataInterface {

    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 1.数据转换map
         * 2.封装样例类数据
         * 3.加载redis涨跌幅数据
         * 4.模式匹配
         * 5.获取匹配模式流数据
         * 6.查询数据
         * 7.发送告警邮件
         */
        //todo 1.数据转换map
        SingleOutputStreamOperator<WarnBaseBean> mapData = waterData.map(new MapFunction<CleanBean, WarnBaseBean>() {
            @Override
            public WarnBaseBean map(CleanBean cleanBean) throws Exception {
                WarnBaseBean warnBaseBean = new WarnBaseBean(
                        cleanBean.getSecCode(),
                        cleanBean.getPreClosePrice(),
                        new BigDecimal(0),
                        new BigDecimal(0),
                        cleanBean.getTradePrice(),
                        cleanBean.getEventTime()
                );
                return warnBaseBean;
            }
        });

        //todo 3.加载redis涨跌幅数据
        JedisCluster jedisCluster = RedisUtil.getJedisCluster();

        //涨跌幅-跌幅 跌幅阈值为负值
        BigDecimal up1 = new BigDecimal(jedisCluster.hget("quot", "upDown1"));
        //涨跌幅-涨幅 涨幅阈值
        BigDecimal up2 = new BigDecimal(jedisCluster.hget("quot", "upDown2"));
        //todo 4.模式匹配
        Pattern<WarnBaseBean, WarnBaseBean> pattern = Pattern.<WarnBaseBean>begin("begin")
                .where(new SimpleCondition<WarnBaseBean>() {
                    @Override
                    public boolean filter(WarnBaseBean value) throws Exception {
                        //涨跌幅=(期末收盘点位-期初前收盘点位)/期初前收盘点位*100%
                        BigDecimal upDownRate = (value.getClosePrice().subtract(value.getPreClosePrice())).divide(value.getPreClosePrice(), 2, RoundingMode.HALF_UP);
                        if (upDownRate.compareTo(up1) == 1 && upDownRate.compareTo(up2) == -1) { //跌幅阈值<涨跌幅<涨幅阈值
                            return false; //涨跌正常数据
                        }
                        return true;
                    }
                });
        //todo 5.获取匹配模式流数据
        PatternStream<WarnBaseBean> cep = CEP.pattern(mapData.keyBy(WarnBaseBean::getSecCode), pattern);
        cep.select(new PatternSelectFunction<WarnBaseBean, Object>() {
            @Override
            public Object select(Map<String, List<WarnBaseBean>> map) throws Exception {
                //todo 6.查询模式匹配成功的数据
                List<WarnBaseBean> begin = map.get("begin");
                //todo 7.发送告警邮件
                if(begin!=null && begin.size()>0){
                    MailSend.send(begin.toString());
                }
                return begin;
            }
        }).print("涨跌幅异常数据>>>");


    }
}
